package com.gaofans.nrpc.client.handler;

import com.gaofans.nrpc.common.processor.EventListener;
import com.gaofans.nrpc.common.processor.EventProcessor;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import com.github.benmanes.caffeine.cache.Scheduler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * @author GaoFans
 */
public class ResponseEventProcessor implements EventProcessor<Object,Long> {

    private final static Logger LOGGER = LoggerFactory.getLogger(ResponseEventProcessor.class);

    private final Cache<Long,EventListener<Object>> listeners;
    private final ReentrantReadWriteLock.ReadLock r;
    private final ReentrantReadWriteLock.WriteLock w;

    public ResponseEventProcessor(Duration expireTime, long maximumSize) {
        this.listeners = Caffeine.newBuilder()
                .expireAfterAccess(expireTime)
                .maximumSize(maximumSize)
                .removalListener((key, graph, cause) -> LOGGER.debug("Listener {} was unregistered {}", key, cause))
                .scheduler(Scheduler.systemScheduler())
                .build();
        ReentrantReadWriteLock lock = new ReentrantReadWriteLock(true);
        this.r = lock.readLock();
        this.w = lock.writeLock();
    }

    @Override
    public void register(Long id, EventListener<Object> listener) {
        r.lock();
        try {
            listeners.put(id,listener);
            LOGGER.debug("Listener {} was registered",id);
        }finally {
            r.unlock();
        }
    }

    @Override
    public EventListener<Object> unRegister(Long id) {
        w.lock();
        try {
            EventListener<Object> listener = listeners.getIfPresent(id);
            listeners.invalidate(id);
            return listener;
        }finally {
            w.unlock();
        }
    }

    @Override
    public EventListener<Object> get(Long id) {
        r.lock();
        try {
            return listeners.getIfPresent(id);
        }finally {
            r.unlock();
        }

    }
}
